package com.aotain.nyx.abnormal;

import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.math3.distribution.NormalDistribution;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.apollo.DosSliding;
import com.aotain.apollo.SlidingWindow;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.common.CommonDB;
import com.aotain.common.mongo.DataColumn;
import com.aotain.nyx.statis.AbnStatisTuple;

@Deprecated
public class AbnormalReverseCheckImport implements FlatMapFunction<Tuple2<String,AbnormalStatTuple>, AbnStatisTuple> {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = -6636160655852119780L;

	private HashMap<String,Double> sipPInfo = new HashMap<String,Double>();//各个目标Ip的熵值
	private HashMap<String,Long> sipPVMap = new HashMap<String,Long>();//目标IP PV数
	private HashMap<String,Long> sdPVMap = new HashMap<String,Long>();//各个源ip到目标ip的PV数
	private Map<String,Long> tripleMap = new HashMap<String,Long>();//各个元组的PV
	private Map<String,Set<String>> sipToTriple = new HashMap<String,Set<String>>();//目标ip下的三元组映射
	private Map<String,Long> delaytimeMap = new HashMap<String,Long>();//求延时Map
	private Map<String,Long> maxDelayMap = new HashMap<String,Long>();//最大延时Map
	private Map<String,Long> minDelayMap = new HashMap<String,Long>();//最小延时Map
	private Map<String,Long> streamPacketMap = new HashMap<String,Long>();//
	private Map<String,Long> streamOctetsMap = new HashMap<String,Long>();//
	private Map<String,Integer> delayTimePVMap = new HashMap<String,Integer>();//
	private HashMap<String,Set<String>> rowKeyMap = new HashMap<String,Set<String>>();//

	private HashMap<String,Long> abnStartTime = new HashMap<String,Long>(); //记录异常开始（第1分钟）
	private HashMap<String,Long> abnUpdateTime = new HashMap<String,Long>(); //DIP 异常追踪时间

	//DIP异常状态，1：发现异常等待持续跟踪，2：确认异常状态 
	private HashMap<String,Integer> abnDIPStatus = new HashMap<String,Integer>();

	//DIP+SIP 入库更新时间
	private HashMap<String,Long> abnRowKeyUpdate = new HashMap<String,Long>(); 

	//一个连接的异常开始时间
	private HashMap<String,String> abnRowKeyStart = new HashMap<String,String>(); 

	//当发现异常，并且启动入库状态是，将当前区间的流量做完成的统计
	private Map<String,Long> memStreamPacketMap = new HashMap<String,Long>();//
	private Map<String,Long> memStreamOctetsMap = new HashMap<String,Long>();//

	String dbJson = null;
	private SlidingWindow cache = new SlidingWindow(20);
	private DosSliding dipPvCache = new DosSliding(20);;

	/**
	 * 攻击峰值
	 */
	private HashMap<String,Double> mapHighFlow = new HashMap<String,Double>();

	@Override
	public void flatMap(Tuple2<String, AbnormalStatTuple> tuple,
			Collector<AbnStatisTuple> out) throws Exception {
		sipPVMap = tuple.f1.getSipPVMap();
		sdPVMap = tuple.f1.getSdPVMap();
		tripleMap = tuple.f1.getTripleMap();
		sipToTriple = tuple.f1.getSipToTriple();
		delaytimeMap = tuple.f1.getDelaytimeMap();
		maxDelayMap = tuple.f1.getMaxDelayMap();
		minDelayMap = tuple.f1.getMinDelayMap();
		streamPacketMap = tuple.f1.getStreamPacketMap();
		streamOctetsMap = tuple.f1.getStreamOctetsMap();
		delayTimePVMap = tuple.f1.getDelayTimePVMap();

		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_m1  = new SimpleDateFormat("yyyyMMddHHmm00");
		SimpleDateFormat sdf_m2  = new SimpleDateFormat("yyyy-MM-dd HH:mm:00");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");  
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		//		sdpvCache.addPInfo(sdPVMap);
		dipPvCache.addPInfo(sipPVMap);
		//求熵
		for(Entry<String,Set<String>> entry:sipToTriple.entrySet()) {
			String sip_idc_dp = entry.getKey();
			Set<String> tripleSet = entry.getValue();
			Long sipPV = sipPVMap.get(sip_idc_dp);
			if(sipPV == 0) {
				//如果概率为0，说明值为0，不计算信息熵，直接返回
				continue;
			}
			double pInfo = 0.0;
			for(Iterator<String> iter=tripleSet.iterator();iter.hasNext();) {
				String triple = iter.next();

				if(tripleMap.get(triple) == 0) {
					continue;
				}

				Double p = (double)tripleMap.get(triple)/(double)sipPV;

				if(p.isNaN()) {
					Logger.getLogger(AbnormalReverseCheckImport.class)
					.info(String.format("###ABN LOG, Check tripleMap is NaN,key:%s value:%d dipPV:%d", triple,tripleMap.get(triple),sipPV));
				}
				//如果p=0 计算信息熵的方法就无法执行 结果为NaN
				Double tmpInfo = -Math.log(p)*p;
				pInfo += tmpInfo;
			}
			if(pInfo != 0.0) {
				sipPInfo.put(sip_idc_dp, pInfo);
			} else {
				//如果有一个阶段信息熵为0，清除窗口
				cache.clearZeroP(sip_idc_dp);
			}
		}

		cache.addPInfo(sipPInfo);
		Map<Integer,HashMap<String,Double>> infoMap = cache.getTuplesWindow();//得到滑动窗口中关于熵的数据
		Map<String,Integer> pInfoLengthMap = new HashMap<String,Integer>();//求熵值平均值时的长度
		Map<String,Double> totalP = new HashMap<String,Double>();//当前所有窗口对应目标元组的熵值和的Map

		if(infoMap != null ) {
			for(HashMap<String,Double> singleMap: infoMap.values()) {
				if(singleMap != null) {
					for(Entry<String,Double> entry : singleMap.entrySet()) {
						String sip_idc_dp = entry.getKey();
						Double value = entry.getValue(); //信息熵
						Double totalPvalue = 0.0;
						if(totalP.containsKey(sip_idc_dp)) {
							totalPvalue = totalP.get(sip_idc_dp);
						}

						//						if(value != 0) {
						Integer sum = pInfoLengthMap.get(sip_idc_dp);
						if(sum == null) {
							sum = 0;
						}
						sum ++;
						pInfoLengthMap.put(sip_idc_dp, sum);
						if(totalPvalue == null || totalPvalue.isNaN()) {
							totalPvalue = value;
						} else {
							totalPvalue += value;
						}
						totalP.put(sip_idc_dp, totalPvalue);
					}
				}
			}

		}

		Map<String,Double> avgP = new HashMap<String,Double>();//目标元组熵值的均值
		for(String dkey : totalP.keySet()) {
			Double avgPvalue = totalP.get(dkey) / (double) pInfoLengthMap.get(dkey);
			if(avgPvalue.isNaN()) {
				Logger.getLogger(AbnormalReverseCheckImport.class)
				.info(String.format("###ABN LOG, Check avgPvalue totalP [%f],pInfoLengthMap:[%d],Key:%s",
						totalP.get(dkey),pInfoLengthMap.get(dkey),dkey));
			}
			avgP.put(dkey, avgPvalue);
		}

		HashMap<String,Double> subsqrMap = new HashMap<String,Double>();
		for(String key : avgP.keySet()) {
			Double value = 0.0;
			if(sipPInfo.get(key) != null && avgP.get(key)!= null) {
				value = (sipPInfo.get(key)-avgP.get(key))*(sipPInfo.get(key)-avgP.get(key));
				if(value.isNaN()) {
					Logger.getLogger(AbnormalReverseCheckImport.class)
					.info(String.format("###ABN LOG, Check Pinfo[%f],avgP:[%f],Key:%s",sipPInfo.get(key),avgP.get(key),key));
				}
			}
			subsqrMap.put(key, value);
		}
		cache.addSubsqr(subsqrMap);

		Map<String,Double> totalSubsqr = new HashMap<String,Double>();//差的平方和汇总
		//		Map<String,Double> stdMap = new HashMap<String,Double>();//标准差集合
		Map<Integer,HashMap<String,Double>> winSums = cache.getSumWindow();
		//		Map<Integer,HashMap<String,Long>> abnNumMaps = sdpvCache.getTuplesWindow();


		for(HashMap<String,Double> singleMap: winSums.values()) {
			if(singleMap != null) {
				for(Entry<String,Double> entry : singleMap.entrySet()) {
					String key = entry.getKey();
					Double value = entry.getValue();
					Double totalSvalue = 0.0;

					if(totalSubsqr.containsKey(key)) {
						totalSvalue = totalSubsqr.get(key);
					}

					if(value.isNaN()) {
						value = 0.0;//后续计算都出现NaN的情况，检测是否由于value = NaN造成。
					}

					if(totalSvalue==null || totalSvalue.isNaN()) {
						totalSvalue = 0.0;
					} else {
						totalSvalue += value;
					}
					totalSubsqr.put(key, totalSvalue);
				}
			}
		}



		Date date = new Date();
		Timestamp dateStr_m = new Timestamp(sdf_m2.parse(sdf_m2.format(date)).getTime());
		String dateStr = sdf.format(date);
		//*********Modified 2017-06-20
		//连续三分钟是否都在置信区间内
		for(String key : totalSubsqr.keySet()) {
			if(pInfoLengthMap.get(key) != null && pInfoLengthMap.get(key) > 15) {//至少满足连续累积数值超过15个时间窗再做异常判断

				if(totalSubsqr.get(key).isNaN()) {//Turk 2017-07-10
					Logger.getLogger(AbnormalReverseCheckImport.class).info("###ABN LOG, totalSubsqr is NaN. Key:" + key);
					continue;
				}

				double stdValue = 0.0;
				double sxn = 0.0;
				double confidenceValueMax = 0.0;//置信区间上界
				double confidenceValueMin = 0.0;//置信区间下界
				double nordis = 0.0;
				double avg = 0.0;
				double sipP = 0.0;
				NormalDistribution s = new NormalDistribution (0,1);//求正态分布分位数所用

				stdValue = Math.sqrt(totalSubsqr.get(key)/((double)pInfoLengthMap.get(key)-(double)1));//S_n
				sxn = stdValue / Math.sqrt(pInfoLengthMap.get(key));//S_xn=S_n/√n
				nordis = s.cumulativeProbability(sxn);
				avg = (avgP.get(key)==null || avgP.get(key).isNaN())?0.0:avgP.get(key);
				sipP = (sipPInfo.get(key)==null||sipPInfo.get(key).isNaN())?0.0:sipPInfo.get(key);
				confidenceValueMax = avg + nordis;
				confidenceValueMin = avg - nordis;

				if(sipP >=confidenceValueMin && sipP <= confidenceValueMax)
				{
					//如果存在正在检测异常的DIP，这个时候将状态设置为0，表示当前分钟没有异常。
					if(abnDIPStatus.containsKey(key)) {
						abnDIPStatus.put(key,0);
						//Logger.getLogger(AbnormalCheckImport.class).info("###ABN LOG, 中断,状态设置为0.KEY:" + key);
					}
				} else {//超出置信区间，开始记录异常行为

					//##########Turk 2017-07-10 #########
					if(!abnStartTime.containsKey(key) || abnDIPStatus.get(key) == 0) {
						//第一次发现,或者中断后又发生
						abnStartTime.put(key,System.currentTimeMillis());
						//更新状态
						abnDIPStatus.put(key, 1);
						abnUpdateTime.put(key, System.currentTimeMillis());
					} else {
						//已记录发现，判断是否持续
						if(abnDIPStatus.get(key) != 2 
								&& System.currentTimeMillis() - abnStartTime.get(key) >= 3*60*1000 - 5*1000L) {
							//不在入库状态，并且开始时间距离当前时间大于2分55秒，说明行为持续，开始入库异常日志
							abnDIPStatus.put(key, 2);

							Logger.getLogger(AbnormalReverseCheckImport.class)
							.info(String.format("###ABN LOG, DIP[%s] 检测到持续异常.[%f-%f] {%f}"
									+ "[stdValue:%f sxn:%f nordis:%f avg:%f] {CK:%f/%d}", key,
									confidenceValueMin,confidenceValueMax,sipP,stdValue,sxn,nordis,avg,
									totalSubsqr.get(key),pInfoLengthMap.get(key)));

						} else {
							//未到达门限时间，继续等待
						}

						if(abnDIPStatus.get(key) == 2) { //更新异常持续时间
							abnUpdateTime.put(key, System.currentTimeMillis());
						}

					}
				}
			} 
		}

		//公式是：S_n=√(1/(n-1) ∑_(i=1)^n?〖(x_i-x)〗^2 ) 、S_xn=S_n/√n
		for(String key : totalSubsqr.keySet()) {

			String sip = key.split("\\*",-1)[0];
			String idcno = key.split("\\*",-1)[1];
			String dport = key.split("\\*",-1)[2];

			if(pInfoLengthMap.get(key) != null && pInfoLengthMap.get(key) >1) {
				//				Date date = new Date();
				//				stdValue = Math.sqrt(totalSubsqr.get(key)/(pInfoLengthMap.get(key)-1));//S_n
				//				sxn = stdValue / Math.sqrt(pInfoLengthMap.get(key));//S_xn=S_n/√n
				//				nordis = s.cumulativeProbability(sxn);
				//				avg = avgP.get(key)==null?0.0:avgP.get(key);
				//				dipP = dipPInfo.get(key)==null?0.0:dipPInfo.get(key);
				//				confidenceValueMax = avg + nordis; 	
				//				confidenceValueMin = avg - nordis;

				Set<String> tripleSet = sipToTriple.get(key);//DIP中当前分钟的所有SIP集合

				if(tripleSet != null) {
					boolean isNormal = true;

					/* Turk 2017-07-10
					 * 1、DIP 发现持续异常，开启允许入库的窗口，在这个窗口内的SIP 都记录入库
					 * 2、新发现的SIP，在开始或者持续中，都以这个SIP第一次出现的那一分钟作为开始时间
					 * 3、在判定DIP 异常结束的时候，将这个阶段入库的SIP 全都设置为结束
					 * 4、在DIP 异常持续中，如果SIP 未持续第二分钟，则在当前分钟结束
					 */
					//$$$$Modified by turk 2016-01-04 ----END

					Set<String> rowKeySet = rowKeyMap.get(key)==null?new HashSet<String>():rowKeyMap.get(key);
					boolean check = true;//信息熵入库校验所需 Modified 2017-06-06
					for(Iterator<String> iter=tripleSet.iterator();iter.hasNext();) {

						String triple = iter.next();
						long packets = streamPacketMap.get(triple);
						String score = Score(packets);
						if(score.equals("100")) {
							continue;
						}

						String[] splits = triple.split("\\*",-1);
						String destport = splits[1];
						String gis = splits[3];
						String protoType = splits[4];
						String dip = splits[0];
						String sdIP = dip + "*" + idcno + "*" + sip;

						String[] gisArray = gis.split("#",-1);
						String destAreaName = gisArray[0];
						String destGis = gisArray[1];
						String sourceAreaName = gisArray[2];//市
						String sourceGis = gisArray[3];
						String sourceAreaCountry = gisArray[4];
						String sourceAreaId = gisArray[5];
						String areaName = gisArray[6];//省
						String sourceAreaCityId = gisArray[7];
						String sourceAreaProvinceId = gisArray[8];
						String evaluate = "";
						String desc = "";
						String abnormal = "0";
						String delayTime = "";
						String rowkey_ = "";
						long attnum = 0;
						String avgdelay = String.format("%d", delaytimeMap.get(sdIP)/delayTimePVMap.get(sdIP));
						abnormal = "1";
						delayTime = "平均延迟："+ avgdelay + "s;最大延时：" + maxDelayMap.get(sdIP) + "s;最小延时：" + minDelayMap.get(sdIP)+"s;";
						attnum = sdPVMap.get(sdIP);
						//						System.out.println("#######1 key=" + sdIP + "; time: " + delaytimeMap.get(sdIP) + ";pv: " + delayTimePVMap.get(sdIP) + "; delayTime" + delayTime);



						String accessTimeKey = dip + "*" + sip + "*" + destport;
						String rowKeyAbnormal = sip + "_" + dateStr + "_" + dip + "_" + destport + "_" + 1;
						if(abnDIPStatus.get(key) != null) {
							//记录攻击开始时间，由于入库的时候是以源IP为粒度入库，所以开始时间记录为当前时间							
							if(abnRowKeyStart.containsKey(accessTimeKey)) {
								dateStr = abnRowKeyStart.get(accessTimeKey);
							}

							if(abnDIPStatus.get(key) == 2 && abnRowKeyUpdate.get(rowKeyAbnormal) == null) {
								//DIP为确认异常状态，并且当前ROWKEY 未入库时
								Logger.getLogger(AbnormalReverseCheckImport.class).info("###ABN LOG IMPORT,KEY=" + key + " ROWKEY= " + rowKeyAbnormal);
								abnRowKeyUpdate.put(rowKeyAbnormal, System.currentTimeMillis()); //INSERT
								abnRowKeyStart.put(accessTimeKey, dateStr);
								InsertLog(rowKeyAbnormal,triple,score,out);
								isNormal = false;
							} else if (abnRowKeyUpdate.get(rowKeyAbnormal) != null){
								//当前ROWKEY 已入库，持续更新记录
								Logger.getLogger(AbnormalReverseCheckImport.class).info("###ABN LOG UPDATE,KEY=" + key + " ROWKEY= " + rowKeyAbnormal);
								abnRowKeyUpdate.put(rowKeyAbnormal, System.currentTimeMillis()); //UPDATE
								UpdateLog(rowKeyAbnormal,triple,out);
								isNormal = false;
							} else if (abnDIPStatus.get(key) == 0) {
								//异常消除，判断是否超过3分钟，满足条件则结束异常告警
								if(abnRowKeyUpdate.get(rowKeyAbnormal) != null && System.currentTimeMillis() - abnRowKeyUpdate.get(rowKeyAbnormal) >= 3*60*1000L) {
									Logger.getLogger(AbnormalReverseCheckImport.class).info("###ABN LOG END,KEY=" + key 
											+ "END TIME:" + abnRowKeyUpdate.get(rowKeyAbnormal) + "ROWKEY:" + rowKeyAbnormal);
									UpdateLog(rowKeyAbnormal,triple,out);
									EndLog(rowKeyAbnormal);
									abnRowKeyUpdate.remove(rowKeyAbnormal);
									abnRowKeyStart.remove(accessTimeKey);
									mapHighFlow.remove(rowKeyAbnormal);
								}
								isNormal = false;
							}
						} else {
							//DIP 异常以停止，终止其他正在告警的SIP（等待3分钟）
							if(abnRowKeyUpdate.get(rowKeyAbnormal) != null && System.currentTimeMillis() - abnRowKeyUpdate.get(rowKeyAbnormal) >= 3*60*1000L) {
								Logger.getLogger(AbnormalReverseCheckImport.class).info("###ABN LOG END,KEY=" + key 
										+ "END TIME:" + abnRowKeyUpdate.get(rowKeyAbnormal) + "ROWKEY:" + rowKeyAbnormal);
								UpdateLog(rowKeyAbnormal,triple,out);
								EndLog(rowKeyAbnormal);
								abnRowKeyUpdate.remove(rowKeyAbnormal);
								abnRowKeyStart.remove(accessTimeKey);
								mapHighFlow.remove(rowKeyAbnormal);
							}
						}

						if(isNormal) {
							//正常日志入库SDS_IDC_LOG     
							//							Date date2 = new Date();
							//							String dateStr2 = sdf.format(date2);	
							//							rowKeyNormal = dip + "_" + dateStr2 + "_" + sip + "_" + dport + "_" + 0;	
							rowkey_ = dip + "_" + dateStr + "_" + sip + "_" + destport + "_" + 0;	
							evaluate = "100";
							abnormal = "0";
							desc = "连接";
							attnum = sdPVMap.get(sdIP);
							delayTime = "平均延迟："+ avgdelay + "s;最大延时：" + maxDelayMap.get(sdIP) + "s;最小延时：" + minDelayMap.get(sdIP)+"s;";
							//							InsertNormalLog(rowkey_,triple);
							//							collector.emit("ABNORMAL_LOG", new Values(rowkey_,dip,sip,destport,dateStr,sourceAreaName,sourceGis,sourceAreaCountry,destAreaName,destGis,evaluate,abnormal,desc,attnum+"",delayTime ,"0"));
						}

					}



					if(abnUpdateTime.get(key) != null && System.currentTimeMillis() - abnUpdateTime.get(key) >= 3*60*1000L) {
						Logger.getLogger(AbnormalReverseCheckImport.class).info("###ABN LOG END,CLEAN MAP,KEY=" + key 
								+ "END TIME:" + abnUpdateTime.get(key));
						abnUpdateTime.remove(key);

						if (abnDIPStatus.get(key) !=null && abnDIPStatus.get(key) == 0) {
							abnDIPStatus.remove(key);
							abnStartTime.remove(key);
						}
					}




					if(rowKeySet.size()>0) {
						rowKeyMap.put(key, rowKeySet);
					}
					/**
					 * 辅助：近20分钟信息熵入库
					 */
					if(!check) {
						List<String> sqlList = new ArrayList<String>();
						for(Entry<Integer,HashMap<String,Double>>entryP : infoMap.entrySet()) {
							int i = entryP.getKey();
							HashMap<String,Double> infoPMap = entryP.getValue();
							if(infoPMap.containsKey(key)) {
								double dp = infoPMap.get(key);
								String sql = "insert into SDS_INFOP(DIP,REPORTTIME,NUM,PINFO,DPORT) "
										+ "	values('"+ sip +"','" + dateStr_m + "',"+ i +","+dp+","+ dport +" )";
								sqlList.add(sql);
								//									System.out.println("########### PINFO test sql is "+ sql);
							}

						}
						//						try {
						//							CommonDB.executeBatchJson(sqlList, dbJson);
						//						} catch (SQLException e) {
						//							// TODO Auto-generated catch block
						//							e.printStackTrace();
						//							Logger.getLogger("##############AbnormalCalBolt PInfo into oracle Excepetion : "+ e.getMessage());
						//						}
					}
				}
			}

			//#######TURK 2017-07-10##########
			ArrayList<String> removeKey = new ArrayList<String>();
			for(Entry<String,Long> entry : abnRowKeyUpdate.entrySet()) {
				//检查是否有ROWKEY 需要结束状态,如果超过3分钟，存在SIP没有数据更新，则认为结束，开启结束流程
				if(System.currentTimeMillis() - entry.getValue() >= 3*60*1000L) {
					Logger.getLogger(AbnormalReverseCheckImport.class).info("###ABN LOG END,KEY=" + key 
							+ "END TIME:" + abnRowKeyUpdate.get(key) + "ROWKEY:" + entry.getKey());
					removeKey.add(entry.getKey());
					//String rowKeyAbnormal = dip + "_" + dateStr + "_" + sip + "_" + destport + "_" + 1;
					String dip = entry.getKey().split("_",-1)[2];
					sip = entry.getKey().split("_",-1)[0];
					String destport = entry.getKey().split("_",-1)[3];
					String accessTimeKey = dip + "*" + sip + "*" + destport;
					String rowKeyAbnormal = entry.getKey();
					EndLog(rowKeyAbnormal);
					abnRowKeyStart.remove(accessTimeKey);
				}
			}

			for(String remove : removeKey) {
				abnRowKeyUpdate.remove(remove);
			}
			//*******TURK 2017-07-10 *******

			//			stdMap.put(key, stdValue);
		}

		tripleMap.clear();//各元组pv清零
		sipPVMap.clear();//dipPV清零
		sipPInfo.clear();
		sipToTriple.clear();
		sdPVMap.clear();
		delaytimeMap.clear();
		maxDelayMap.clear();
		minDelayMap.clear();
		delayTimePVMap.clear();
		streamPacketMap.clear();
		streamOctetsMap.clear();
		cache.advanceHead();
		//		sdpvCache.advanceHead();
		dipPvCache.advanceHead();

	}


	private void InsertNormalLog(String rowKeyNormal,String rowValue){
		String triple = rowValue;
		String[] splits = triple.split("\\*",-1);
		String destport = splits[1];
		String gis = splits[3];
		String protoType = splits[4];
		String sip = splits[2];
		String dip = splits[0];
		String idcno = splits[5];
		String flowDir = splits[6];
		String sdIP = dip + "*" + idcno + "*" + sip;

		String dateStr = rowKeyNormal.split("_",-1)[1];

		String[] gisArray = gis.split("#",-1);
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];//市
		String sourceGis = gisArray[3];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];
		String areaName = gisArray[6];//省
		String sourceAreaCityId = gisArray[7];
		String sourceAreaProvinceId = gisArray[8];
		String desc = "正常连接";
		String delayTime = "";

		long attnum = 0;
		String avgdelay = String.format("%d", delaytimeMap.get(sdIP)/delayTimePVMap.get(sdIP));
		delayTime = "平均延迟："+ avgdelay + "s;最大延时：" + maxDelayMap.get(sdIP) + "s;最小延时：" + minDelayMap.get(sdIP)+"s;";
		attnum = sdPVMap.get(sdIP);


		String sAbnormalTable = "SDS_ABNORMAL_LOG2";
		MongoImportTool importtool = MongoImportTool.getInstance();
		List<DataColumn> row = new ArrayList<DataColumn>();
		row.add(new DataColumn("ROWKEY", rowKeyNormal)); //主键字段，用于更新记录
		row.add(new DataColumn("SOURCEIP", sip));
		row.add(new DataColumn("DESTPORT", destport));
		row.add(new DataColumn("ACCESSTIME", Long.parseLong(dateStr)));
		row.add(new DataColumn("ABRNORMAL", 0));
		row.add(new DataColumn("DESTIP", dip));
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO", sourceGis));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO", destGis));
		row.add(new DataColumn("ATTNUM", attnum+""));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", "100"));
		row.add(new DataColumn("PROVINCE", areaName));
		row.add(new DataColumn("FLOWDIRECTION", flowDir));
		row.add(new DataColumn("UPSTREAMOCTETS", streamOctetsMap.get(triple)));
		row.add(new DataColumn("UPSTREAMPACKET", streamPacketMap.get(triple)));

		//		System.out.println("NormalLog #######Into MongoDB#####" + row.toString());
		importtool.InsertRowData(sAbnormalTable, row);	

	}

	/**
	 * 新发现的异常连接
	 * @param rowKeyAbnormal
	 * @param row
	 * @param score
	 * @throws Exception
	 * @Description:
	 */
	private void InsertLog(String rowKeyAbnormal, String rowValue,String score ,Collector<AbnStatisTuple> out ) throws Exception {
		//String row = dip + "*" + dport + "*" + sip + "*" + gisstr + "*" + type + "*" + idcno;



		String triple = rowValue;
		String[] splits = triple.split("\\*",-1);
		String destport = splits[1];
		String gis = splits[3];
		String protoType = splits[4];
		String sip = splits[2];
		String dip = splits[0];
		String idcno = splits[5];
		String flowDir = splits[6];
		String sdIP = dip + "*" + idcno + "*" + sip;

		String[] gisArray = gis.split("#",-1);
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];//市
		String sourceGis = gisArray[3];
		String destAreaCountry = gisArray[4];
		String destAreaId = gisArray[5];
		String areaName = gisArray[6];//省
		String destAreaCityId = gisArray[7];
		String destAreaProvinceId = gisArray[8];
		String sourceAreaCountry = gisArray[9];
		String evaluate = score;
		//		String abnormal = "1";
		String delayTime = "";

		double maxValue =  streamOctetsMap.get(triple)/(1024*60);
		mapHighFlow.put(rowKeyAbnormal, maxValue);
		String desc = String.format("异常连接, 峰值流量[%f KB/s]", maxValue);

		long attnum = 0;
		String avgdelay = String.format("%d", delaytimeMap.get(sdIP)/delayTimePVMap.get(sdIP));
		delayTime = "平均延迟："+ avgdelay + "s;最大延时：" + maxDelayMap.get(sdIP) + "s;最小延时：" + minDelayMap.get(sdIP)+"s;";
		attnum = sdPVMap.get(sdIP);

		String accessTimeKey = dip + "*" + sip + "*" + destport;

		if(!abnRowKeyStart.containsKey(accessTimeKey))
			return;

		String dateStr =  abnRowKeyStart.get(accessTimeKey);

		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");  
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");

		Date dtTime = sdf.parse(dateStr);

		String dateStr_h = sdf_h.format(dtTime);
		String dateStr_d = sdf_d.format(dtTime);
		String dateStr_hour = sdf_hour.format(dtTime);
		String sAbnormalTable = "SDS_ABNORMAL_LOG2";
		MongoImportTool importtool = MongoImportTool.getInstance();
		List<DataColumn> row = new ArrayList<DataColumn>();
		row.add(new DataColumn("ROWKEY", rowKeyAbnormal)); //主键字段，用于更新记录
		row.add(new DataColumn("SOURCEIP", sip));
		row.add(new DataColumn("DESTPORT", destport));
		row.add(new DataColumn("ACCESSTIME", Long.parseLong(dateStr)));
		row.add(new DataColumn("ABRNORMAL", 1));
		row.add(new DataColumn("DESTIP", dip));
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO", sourceGis));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO", destGis));
		row.add(new DataColumn("ATTNUM", attnum+""));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", "40"));
		row.add(new DataColumn("PROVINCE", areaName));
		row.add(new DataColumn("DESTCOUNTRY", destAreaCountry));
		row.add(new DataColumn("FLOWDIRECTION", flowDir));
		row.add(new DataColumn("UPSTREAMOCTETS", streamOctetsMap.get(triple)));
		row.add(new DataColumn("UPSTREAMPACKET", streamPacketMap.get(triple)));
		row.add(new DataColumn("ATTTYPE","1"));
		importtool.InsertRowData(sAbnormalTable, row);	

		AbnStatisTuple record = new AbnStatisTuple();
		record.setDestIP(dip);
		record.setSourceIP(sip);
		record.setType("REQUESTABNL");
		record.setAttackNum(1l);
		record.setStAttackNum(1l);
		record.setUpStreamOctets(streamOctetsMap.get(triple));
		record.setUpStreamPacket(streamPacketMap.get(triple));
		record.setAttackType("1");
		record.setAttackMax(maxValue);
		record.setAttTime(dateStr);
		out.collect(record);
		memStreamOctetsMap.put(rowKeyAbnormal, streamOctetsMap.get(triple));
		memStreamPacketMap.put(rowKeyAbnormal, streamPacketMap.get(triple));

	}

	/**
	 * 已发现的异常连接，更新字节和数据包值
	 * @param rowKeyAbnormal
	 * @param row
	 * @throws Exception
	 * @Description:
	 */
	private void UpdateLog(String rowKeyAbnormal, String rowValue,Collector<AbnStatisTuple> out ) throws Exception {

		String triple = rowValue;
		String[] splits = triple.split("\\*",-1);
		String destport = splits[1];
		String gis = splits[3];
		String protoType = splits[4];
		String sip = splits[2];
		String dip = splits[0];
		String idcno = splits[5];
		String sdIP = dip + "*" + idcno + "*" + sip;

		String accessTimeKey = dip + "*" + sip + "*" + destport;

		if(!abnRowKeyStart.containsKey(accessTimeKey))
			return;

		String dateStr =  abnRowKeyStart.get(accessTimeKey);

		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");  
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");

		Date dtTime = sdf.parse(dateStr);

		String dateStr_h = sdf_h.format(dtTime);
		String dateStr_d = sdf_d.format(dtTime);
		String dateStr_hour = sdf_hour.format(dtTime);


		String[] gisArray = gis.split("#",-1);
		String sourceAreaName = gisArray[2];//市
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];
		String areaName = gisArray[6];//省


		if(memStreamOctetsMap.containsKey(rowKeyAbnormal)) {
			memStreamOctetsMap.put(rowKeyAbnormal, memStreamOctetsMap.get(rowKeyAbnormal) 
					+ streamOctetsMap.get(triple));
		}

		if(memStreamPacketMap.containsKey(rowKeyAbnormal)) {
			memStreamPacketMap.put(rowKeyAbnormal, memStreamPacketMap.get(rowKeyAbnormal) 
					+ streamPacketMap.get(triple));
		}

		double maxValue = (double)streamOctetsMap.get(triple)/(double)60/(double)1024 ;

		if(mapHighFlow.containsKey(rowKeyAbnormal))
		{
			double maxValueMem = mapHighFlow.get(rowKeyAbnormal);
			if(maxValueMem < maxValue)
			{
				maxValueMem = maxValue;
				mapHighFlow.put(rowKeyAbnormal, maxValueMem);
			}
			else
			{
				maxValue = maxValueMem;
			}
		}
		else
		{
			mapHighFlow.put(rowKeyAbnormal, maxValue);
		}
		String desc = String.format("异常连接, 峰值流量[%f KB/s]", maxValue);

		String sAbnormalTable = "SDS_ABNORMAL_LOG2";
		MongoImportTool importtool = MongoImportTool.getInstance();
		List<DataColumn> row = new ArrayList<DataColumn>();

		row.add(new DataColumn("UPSTREAMOCTETS", memStreamOctetsMap.get(rowKeyAbnormal)));
		row.add(new DataColumn("UPSTREAMPACKET", memStreamPacketMap.get(rowKeyAbnormal)));
		row.add(new DataColumn("DESC", desc));
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(sAbnormalTable, filter, row);

		AbnStatisTuple record = new AbnStatisTuple();
		record.setDestIP(dip);
		record.setSourceIP(sip);
		record.setType("SESSIONMIDDLE");
		record.setUpStreamOctets(streamOctetsMap.get(triple));
		record.setUpStreamPacket(streamPacketMap.get(triple));
		record.setAttackType("1");
		record.setAttackMax(maxValue);
		record.setAttTime(dateStr);
		out.collect(record);

		//		collector.emit(new Values(dip,sip,destport,"",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,areaName,
		//				0l,dateStr_hour,"SESSIONMIDDLE",idcno,sourceAreaCountry,Long.valueOf(streamOctetsMap.get(triple)),Long.valueOf(streamPacketMap.get(triple))));


	}

	/**
	 * 判断连接消失后，更新结束时间
	 * @param rowKeyAbnormal
	 * @Description:
	 */
	private void EndLog(String rowKeyAbnormal) {

		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		String endtime = sdf.format(abnRowKeyUpdate.get(rowKeyAbnormal) + 60*1000L);

		String sAbnormalTable = "SDS_ABNORMAL_LOG2";
		MongoImportTool importtool = MongoImportTool.getInstance();
		List<DataColumn> row = new ArrayList<DataColumn>();

		row.add(new DataColumn("ENDTIME", Long.parseLong(endtime)));
		importtool.InsertRowData(sAbnormalTable, row);	

		//清理作案现场
		memStreamOctetsMap.remove(rowKeyAbnormal);
		memStreamPacketMap.remove(rowKeyAbnormal);

	}

	/**
	 * 
	 * @param abnNum 每分钟累积包数
	 * @return
	 * @Description:
	 */
	private String Score(long abnNum) {
		String score = "100";
		if(abnNum <= 50 ) {
			score = "100";
		} else if( abnNum > 50 && abnNum <= 100) {
			score = "60";
		} else if( abnNum > 200) {
			score = "40";
		}
		return score;
	}

	/**
	 * 
	 *  检查各个内存中MAP的大小，为清理内存提供依据
	 * @Description:
	 */
	private void CheckMapSize() {
		Logger.getLogger(AbnormalReverseCheckImport.class).info("##HM abnStartTime=" + abnStartTime.size());
		Logger.getLogger(AbnormalReverseCheckImport.class).info("##HM abnConfirm=" + abnUpdateTime.size());
		Logger.getLogger(AbnormalReverseCheckImport.class).info("##HM abnDIPStatus=" + abnDIPStatus.size());
		Logger.getLogger(AbnormalReverseCheckImport.class).info("##HM abnRowKeyUpdate=" + abnRowKeyUpdate.size());
		Logger.getLogger(AbnormalReverseCheckImport.class).info("##HM memStreamPacketMap=" + memStreamPacketMap.size());
		Logger.getLogger(AbnormalReverseCheckImport.class).info("##HM memStreamOctetsMap=" + memStreamOctetsMap.size());

	}


}
